package cn.itedus.lottery.application.mq.producer;

import cn.itedus.lottery.common.Constants;
import cn.itedus.lottery.domain.activity.model.vo.ActivityPartakeRecordVO;
import cn.itedus.lottery.domain.activity.model.vo.InvoiceVO;
import cn.itedus.lottery.domain.activity.service.partake.IActivityPartake;
import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;


import javax.annotation.Resource;

/**
 * @author aodre , QQ : 480029069
 * @date 2023/10/5 16:04
 */
@Component
public class RocketMQProducer {

    private Logger logger = LoggerFactory.getLogger(KafkaProducer.class);

    @Resource
    RocketMQTemplate rocketMQTemplate;

    @Resource
    private IActivityPartake activityPartake;


    public static final String TOPIC_INVOICE = "lottery_invoice";

    /**
     * MQ主题：活动领取记录
     */
    public static final String TOPIC_ACTIVITY_PARTAKE = "lottery_activity_partake";

    /**
     * 发送中奖物品发货单消息
     *
     * @param invoice 发货单
     */
    public void sendLotteryInvoice(InvoiceVO invoice) {
        String objJson = JSON.toJSONString(invoice);
        logger.info("发送MQ消息(中奖发货单) topic：{} bizId：{} message：{}", TOPIC_INVOICE, invoice.getuId(), objJson);
        rocketMQTemplate.asyncSend(TOPIC_INVOICE, objJson, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                // 发送成功， 直接更新 mq_state = 1
                // 4.1 MQ 消息发送完成，更新数据库表 user_strategy_export.mq_state = 1
                activityPartake.updateInvoiceMqState(invoice.getuId(), invoice.getOrderId(), Constants.MQState.COMPLETE.getCode());
            }

            @Override
            public void onException(Throwable throwable) {
                // 发送失败更新 mq_state = 2 ,标识失败, 后续由 worker 扫描补偿
                // 4.2 MQ 消息发送失败，更新数据库表 user_strategy_export.mq_state = 2 【等待定时任务扫码补偿MQ消息】
                activityPartake.updateInvoiceMqState(invoice.getuId(), invoice.getOrderId(), Constants.MQState.FAIL.getCode());
            }
        });
    }

    /**
     * 发送领取活动记录MQ
     * 用来实现 ,  redis 库存 和 mysql库存 最终的一致性
     *
     * @param activityPartakeRecord 领取活动记录
     */
    public void sendLotteryActivityPartakeRecord(ActivityPartakeRecordVO activityPartakeRecord) {
        String objJson = JSON.toJSONString(activityPartakeRecord);
        logger.info("发送MQ消息(领取活动记录) topic：{} bizId：{} message：{}", TOPIC_ACTIVITY_PARTAKE, activityPartakeRecord.getuId(), objJson);
//        return kafkaTemplate.send(TOPIC_ACTIVITY_PARTAKE, objJson);
        rocketMQTemplate.asyncSend(TOPIC_ACTIVITY_PARTAKE, objJson, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                // 成功的话, 其实不需要做什么
                //TODO  记个 log
            }

            @Override
            public void onException(Throwable throwable) {
                // 失败的话, 需要做的逻辑, 比如继续重试, 还是记日志 ,  还是 入库
                logger.info("发送MQ消息(领取活动记录) topic：{} bizId：{} message：{} 失败,请注意!", TOPIC_ACTIVITY_PARTAKE, activityPartakeRecord.getuId(), objJson);
                logger.info("具体的细节  ----->  {}",throwable.toString());
                // TODO 补偿的逻辑
            }
        });
    }
}
